package com.learning.netease.distributed.lock.zk.zklock.improve;

import com.learning.netease.distributed.lock.zk.zklock.MyZkSerializer;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * ClassName: ZkDistributedImproveLock
 * Description: 基于ZK实现的分布式锁
 * <p>
 * 基于临时有序节点获取实现，只关心前一个节点
 * <p>
 * Date: 2019/3/20 14:19 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
public class ZkDistributedImproveLock implements Lock {


    /**
     * 利用临时有序节点来实现分布式锁
     */
    private String lockPath;

    private ZkClient client;

    private String currentPath;

    private String beforePath;

    public ZkDistributedImproveLock(String host, String lockPath) {
        super();
        this.lockPath = lockPath;
        this.client = new ZkClient(host);
        client.setZkSerializer(new MyZkSerializer());

        // 尝试创建父节点
        if (!client.exists(lockPath)) {
            try {
                client.createPersistent(lockPath);
            } catch (ZkNodeExistsException e) {

            }
        }
    }

    @Override
    public boolean tryLock() {
        if (currentPath == null) {
            // 临时有序节点，[0000000000] 这种格式
            currentPath = client.createEphemeralSequential(lockPath + "/", "lock");
        }

        // 获取所有子节点
        List<String> children = client.getChildren(lockPath);

        // 排序
        Collections.sort(children);

        // 判断当前节点是否是最小的
        if (currentPath.endsWith(lockPath + "/" + children.get(0))) {
            System.out.println("当前节点: " + currentPath);
            return true;
        } else {
            // 获取前一个节点
            // 获取字节的索引
            int curIndex = children.indexOf(currentPath.substring(lockPath.length() + 1));
            beforePath = lockPath + "/" + children.get(curIndex - 1);
            System.out.println("前一个节点: " + beforePath);
        }
        return false;
    }

    @Override
    public void lock() {
        // 获取获取不到锁，则阻塞
        if (!tryLock()) {
            // 阻塞自己，持续获取锁
            waitForLock();
            // 再次尝试
            lock();
        }
    }

    /**
     * 注册watcher、阻塞自己
     */
    private void waitForLock() {
        CountDownLatch countDownLatch = new CountDownLatch(1);

        // 创建一个监听
        IZkDataListener listener = new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {

            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                System.out.println("***** 收到节点被删除了 *****");

                // 节点删除，意味着锁释放了，可以去获取
                countDownLatch.countDown();
            }
        };

        // 注册监听
        client.subscribeDataChanges(beforePath, listener);

        // 阻塞自己，如果节点已经存在则阻塞
        if (client.exists(beforePath)) {
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        // 取消监听，方法退出，后续会继续调用lock方法
        client.unsubscribeDataChanges(beforePath, listener);

    }

    @Override
    public void unlock() {
        client.delete(currentPath);
    }


    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }
}
